package com.gjy.learning.scala.sql

import com.gjy.learning.scala.SparkReadOracle.oracleDFCreate
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{SparkSession, functions}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.IntegerType
import org.scalatest.BeforeAndAfterEach
import org.scalatest.funspec.AnyFunSpec
import org.scalatest.matchers.should.Matchers

import scala.language.postfixOps

class testEasySQL1 extends AnyFunSpec with Matchers with BeforeAndAfterEach {
  val spark: SparkSession = SparkSession.builder()
    .appName("DataType Conversion")
    .master("local[*]")
    .getOrCreate()
  private val emp = oracleDFCreate(spark, "emp")
  private val dept = oracleDFCreate(spark, "dept")
  //  emp.show()

  import com.gjy.learning.scala.util.GetUtil._
  import spark.implicits._

  override def beforeEach(): Unit = {
  }

  override def afterEach(): Unit = {
    spark.stop()
  }

  describe("SparkSQLDSLTest") {
    it("//1查询出员工号为7521，7900，7782的所有员工的信息") {
      emp.filter($"empno" === 7521 || $"empno" === 7900 || $"empno" === 7782).show()
      emp.filter($"empno" isin(7521, 7900, 7782)).show()
    }
    it("2查出emp表中薪水在3000以上（包括3000）的所有员工的员工号、姓名、薪水") {
      emp.filter($"sal" >= 3000).select("empno", "ename", "sal").show()
    }
    it("查询出emp表中部门编号为20，薪水在2000以上（不包括2000）的所有员工，显示他们的员工号，姓名以及薪水，以如下列名显示：员工编号 员工名字 薪水") {
      emp.filter($"deptno" === 20 && $"sal" > 2000)
        .select($"empno".as("员工编号"), $"ename".as("员工名字"), $"sal".as("薪水"))
        .show()
    }
    it("查询出薪水在800到2500之间（闭区间）所有员工的信息") {
      emp.filter($"sal" >= 800 && $"sal" <= 2500).show()
      emp.filter($"sal".between(800, 2500)).show()
    }
    it("列出最低薪金大于1500的各种工作") {
      emp.groupBy($"job")
        .agg(min($"sal").as("min_sal"))
        .filter($"min_sal" > 1500)
        .select($"job").show()
    }
    it("列出在每个部门工作的员工数量、平均工资和平均服务期限") {
      emp.groupBy($"deptno")
        .agg(count(lit(1)).as("员工数量"), avg($"sal").as("平均工资"),
          avg(datediff(current_date(), $"hiredate")).as("平均服务期限"))
        .show
    }
    it("列出各种工作的最低工资") {
      emp.groupBy($"job")
        .agg(min($"sal").as("最低工资"))
        .show
    }
    it("列出各个部门的 MANAGER（经理）的最低薪金。") {
      emp.filter($"job" === "MANAGER")
        .groupBy($"deptno")
        .agg(min($"sal").as("最低薪金"))
        .show
    }
    it("列出所有员工的年工资,按年薪从低到高排序。") {
      emp.select($"empno", $"ename", $"job",
          ($"sal" * 12 + when($"comm".isNull, 0).otherwise($"comm")).as("年薪"))
        .sort($"年薪").show
    }
    it("查询出emp表中所有的工作种类（无重复）") {
      emp.select("job").distinct().show

      emp.groupBy("job")
        .agg(first("job")).select($"job").show

      emp.select("job").dropDuplicates().show
    }
    it("查询出所有奖金（comm）字段不为空的人员的所有信息。") {
      emp.filter($"comm".isNotNull).select("*").show
      emp.filter($"comm" >= 0).select("*").show
    }
    it("将所有员工按薪水升序排序，薪水相同的按照入职时间降序排序。") {
      emp.sort($"sal" asc, $"hiredate" desc).show
    }
    it("查出emp表中所有部门的最高薪水和最低薪水，部门编号为10的部门不显示。") {
      emp.filter($"deptno" =!= 10).groupBy($"deptno")
        .agg(max($"sal").as("最高薪水"), min($"sal").as("最低薪水"))
        .show
    }
    it("列出薪金比“SMITH”多的所有员工。") {
      val smithSalary = emp.filter($"ename" === "SMITH").select($"sal").first().get(0)
      //      print(smithSalary)
      emp.filter($"sal" > smithSalary).select("empno", "ename").show
    }
    it("查询工资 与 SCOTT 相同的员工信息.") {
      val scottSalary = emp.filter($"ename" === "SCOTT").select($"sal").first().get(0)
      emp.filter($"sal" === scottSalary && $"ename" =!= "SCOTT").select("*").show
    }
    it("列出出薪金高于公司平均薪金的所有员工") {
      val avgSalary = emp.groupBy().agg(avg($"sal").as("平均薪金")).first().get(0)
      emp.filter($"sal" > avgSalary).select("ename").show
    }
    it("列出与“SCOTT”从事相同工作的所有员工。1") {
      val scottJob = emp.filter($"ename" === "scott").select($"job")
      val scottJobNotNull =
        if (scottJob.isEmpty) "SCOTT's job is null"
        else scottJob.first().get(0)

      emp.filter($"job" === scottJobNotNull).select("ename").show
    }
    it("列出与“SCOTT”从事相同工作的所有员工。2") {
      val scottJobOption = emp.filter($"ename" === "SCOTT").select($"job").getFirstString
      val scottJobOption1 = emp.filter($"ename" === "SCOTT").select($"job").getHeadOption

      emp.filter($"job" === scottJobOption).select($"ename").filter($"ename" =!= "SCOTT").show()
      emp.filter($"job" === scottJobOption1).select($"ename").filter($"ename" =!= "SCOTT").show()

      /*      val resultDF = scottJobOption match {
              case Some(scottJob) =>
                emp.filter($"job" === scottJob.get(0)).select($"ename").filter($"ename" =!= "SCOTT")
              case None =>
                spark.emptyDataFrame
            }
            resultDF.show()*/
    }
    it("列出薪金等于部门30中员工的薪金的非30号部门的员工的姓名和薪金。") {
      val deptno30Sal = emp.filter($"deptno" === 30)
      val deptnoOtherSal = emp.filter($"deptno" =!= 30)
      deptnoOtherSal.join(deptno30Sal, deptnoOtherSal("sal") === deptno30Sal("sal"))
        .select($"ename", $"sal")
    }
    it("列出薪金等于部门30中员工的薪金的非30号部门的员工的姓名和薪金。2") {
      val deptno30Sal = emp.filter($"DEPTNO" === 30).select($"SAL").collect().map(_.get(0))
      println(s"deptno30Sal 的数据类型是: ${deptno30Sal.getClass}")
      println(deptno30Sal.mkString(","))

      emp.filter(
        $"SAL".isin(deptno30Sal: _*) &&
          $"DEPTNO" =!= 30
      ).select("ENAME", "SAL", "DEPTNO").show
    }
    it("比最高还高的非30部门员工") {
      emp.filter($"DEPTNO" =!= 30)
        .filter($"SAL" === emp.filter($"DEPTNO" === 30).agg(max($"SAL")).getHeadOption)
        .select($"ename", $"deptno", $"sal").show
    }
    it("选择所有员工的姓名、员工号以及他的管理者的姓名和员工号") {
      emp.as("E").join(emp.as("M"), $"E.MGR" === $"M.EMPNO", "leftouter")
        .select($"E.EMPNO",
          $"E.ENAME",
          $"M.EMPNO".as("MGRNO"),
          $"M.EMPNO".as("MGRNAME")).show()
    }
    it("列出受雇日期早于其直接上级的所有员工。") {
      emp.as("E").join(emp.as("M"), $"E.MGR" === $"M.EMPNO", "leftouter")
        .filter($"E.HIREDATE" < $"M.HIREDATE")
        .select($"E.EMPNO",
          $"E.ENAME",
          $"M.EMPNO".as("MGRNO"),
          $"M.EMPNO".as("MGRNAME")).show()
    }
    it("列出部门名称和这些部门的员工信息，同时列出那些没有员工的部门。") {
      emp.as("e").join(dept.as("d"), $"e.deptno" === $"d.deptno", "rightouter")
        .select($"d.dname", $"d.deptno", $"e.empno", $"e.ename")
        .show
    }
    it("列出所有“CLERK”（办事员）的姓名及其部门名称。") {
      emp.as("e").filter($"job" === "CLERK").join(dept.as("d"), $"e.deptno" === $"d.deptno", "leftouter")
        .select($"e.ename", $"d.dname").show()
    }
    it("列出在部门“SALES”（销售部）工作的员工的姓名，假定不知道销售部的部门编号。") {
      emp.as("e").join(dept.as("d").filter($"d.dname" === "SALES"), $"e.deptno" === $"d.deptno", "inner")
        .select($"e.ename", $"d.dname").show()
    }
    it("列出所有员工的姓名、部门名称和工资") {
      emp.as("e").join(dept.as("d"), $"e.deptno" === $"d.deptno", "leftouter")
        .select($"e.ename", $"d.dname", $"e.sal").show()
    }
    it("查询所有部门的信息以及对应的员工数量和工资平均值") {
      emp.as("e").join(dept.as("d"), $"e.deptno" === $"d.deptno", "rightouter")
        .groupBy($"d.deptno", $"d.dname", $"d.loc")
        .agg(count($"empno").as("员工数量"), avg($"e.sal").as("工资平均值"))
        .show()
    }
    it("选择所有有奖金的员工的姓名及其所在的部门信息") {
      emp.as("e").filter($"e.comm" > 0)
        .join(dept.as("d"), $"e.deptno" === $"d.deptno", "leftouter")
        .select($"e.ename", $"d.deptno", $"d.dname", $"d.loc")
        .show
    }
    it("查询员工ENAME的第三个字母是A的员工的信息(使用2种方法)。") {
      emp.filter($"ename".substr(3, 1) === "A").show
      emp.filter($"ename" like "__A%").show
    }
    it("使用trim函数将字符串'hello'、'?Hello?' 分别处理得到下列字符串ello、Hello") {
      //分别获取字符串hello和?Hello?的dataframe类型对象
      val helloDF = Seq("hello").toDF("hello")
      val questionHelloDF = Seq("?Hello?").toDF("questionHello")
      //使用trim函数去掉字符串前后的空格
      helloDF.select(trim($"hello", "h").as("trimHello")).show()
      questionHelloDF.select(trim($"questionHello", "?").as("trimQuestionHello")).show()
    }
    it("查询部门10, 20的员工截止到2000年1月1日，工作了多少个月") {
      emp.filter($"DEPTNO".isin(10, 20))
        .select(
          $"EMPNO",
          $"ENAME",
          months_between(
            to_timestamp(lit("2000-01-1"), "yyyy-MM-dd"),
            $"HIREDATE"
          ).as("months_between")
        ).show
    }
    it("查询所有名字的开始字母是J、A或M的雇员,用首字母大写、其它字母小写显示雇员的全名，显示名字的长度，并对查询结果按雇员的全名升序排序。") {
      emp.filter($"ENAME".substr(1, 1) isin("J", "A", "M"))
        .select(initcap($"ENAME").alias("ENAME"),
          functions.length($"ENAME").alias("LENGTH"))
        .orderBy($"ENAME").show
    }
    it("查询员工姓名中包含大写或小写字母A的员工姓名") {
      emp.filter(upper($"ename").contains("A")).select($"ename").show
    }
    it("查询部门编号为10或20，入职日期在81年5月1日之后，并且姓名中包含大写字母A的员工姓名，员工姓名长度") {
      emp.filter(($"deptno" isin(10, 20)) &&
          ($"hiredate" > to_date(lit("1981-05-01"), "yyyy-MM-dd")) &&
          ($"ename" contains "A"))
        .select($"ename".as("员工姓名"), functions.length($"ename").as("姓名长度"))
        .show()
    }
    it("查询员工的姓名及其经理编号，要求对于没有经理的显示“No Manager”字符串。") {
      emp.select($"ename", when($"mgr".isNull, "No Manager").otherwise($"mgr").as("mgr"))
        .show()
    }
    it("查询员工的姓名及其job，要求对于job不为经理或总裁的的显示“普通员工”，其它显示为“管理者”") {
      emp.select($"ename".alias("员工姓名"),
          when($"job".isin("MANAGER", "PRESIDENT"), "管理者").otherwise("普通员工").alias("员工工作"))
        .show
    }
    it("将员工工资按如下格式显示：123,234.00 RMB。") {
      emp.select($"ename",
          concat(lit("$"), format_number($"sal", 2), lit(" RMB")).as("salary"))
        .show()
    }
    it("在员工表中查询出员工的工资，并计算应交税款：" +
      "如果工资小于1000,税率为0，" +
      "如果工资大于等于1000并小于2000，税率为10％，" +
      "如果工资大于等于2000并小于3000，税率为15％，" +
      "如果工资大于等于3000，税率为20％。") {
      emp.select($"ename",
          $"sal",
          when($"sal" >= 3000, 0.2).when($"sal" >= 2000, 0.15).when($"sal" >= 1000, 0.1).otherwise(0).as("tax_rate"))
        .withColumn("tax", $"sal" * $"tax_rate")
        .show()
    }
    it("显示姓名、hiredate和雇员开始工作日是星期几(提示：使用to_char函数)") {
      emp.select($"ename",
          $"hiredate",
          to_date($"hiredate", "yyyy-MM-dd").as("hiredate1"),
          date_format($"hiredate", "EEEE").as("hiredate2"))
        .show()
    }
    it("需要统计G1220每组（NAME）连续时间（TIME）内的连续完成数（COMPLETE）" +
      "其中有某一时间的完成数为0就重新计算。") {
      val g1220 = oracleDFCreate(spark, "G1220")
      g1220.select($"name", $"time", $"complete")
        .withColumn("flag", $"time" - row_number().over(Window.partitionBy("name", "complete").orderBy("time")))
        .select($"name", $"time", $"complete",
          sum("complete").over(Window.partitionBy("name", "complete", "flag").orderBy("time")).cast(IntegerType).as("连续完成数"))
        .sort("name", "time")
        .show()
    }
    it("有一张充值表T1229，先需要根据财务的需求，根据充值日期、" +
      "\n有效天数和充值金额分摊到2020年最后一天，即2020年12月31日。" +
      "\n  需要得到如下的表，即在上表的基础上增加两列分摊金额和剩余金额。" +
      "\n  分摊金额时，包括充值日期和2020年12月31日这两天，即包括头尾日期。" +
      "\n  解释：例如2020-09-10这天充值了2000元，从2020-09-10到2020-12-31日这一天总共有113天，" +
      "\n  实际有效期为240天，那么到2020-12-31日这一天，" +
      "\n  需要分摊这2000元的金额计算方式为：2000/240*113=941.6629。" +
      "\n  如果有效天数小于到2020-12-31日这天的天数，那么就全部分摊。") {
      val t1229 = oracleDFCreate(spark, "T1229")
      println(getDFScheamaType(t1229, "充值金额"))
      t1229.withColumn("剩余天数", datediff(lit("2020-12-31"), $"充值日期") + 1)
        .withColumn("分摊金额", format_number(when($"剩余天数" < $"有效天数", ($"充值金额" / $"有效天数") * $"剩余天数").otherwise($"充值金额"), 4))
        .withColumn("剩余金额",  format_number(when($"剩余天数" < $"有效天数", $"充值金额" - $"分摊金额").otherwise(0),4))
        .drop("剩余天数")
        .show
    }
  }
}